// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: jing

#include "coin2/exchange/upbit/feed_v1/processor.h"

#include <string>
#include <utility>

#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/upbit/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::upbit::feed_v1::impl {
using namespace std::string_view_literals;

void UpbitFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    THROW() << "JSON parse error: " << GetParseError_En(document.GetParseError());
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (document.HasMember("type")) {
    const std::string& channel = document["type"].GetString();
    if (!CheckChannelFilter(channel)) {
      return;
    }

    const std::string& native_symbol = document["code"].GetString();
    if (channel == "trade") {
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        return;
      }
      Trade* trade = upd_mgr_.GetTradeMutable();
      UpbitFeedUpdateParser::ParseTrade(document, trade);
      if (document["stream_type"].GetString() == "REALTIME"sv){
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    } else if (channel == "orderbook") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        UpbitFeedUpdateParser::ParseBook(book, document);
        if (document["stream_type"].GetString() == "REALTIME"sv){
          upd_mgr_.PublishBook();
        }
      }
    } else {
      LOG(INFO) << "[Upbit] unknown msg for " << packet.topic_name << ": " << packet.payload.data();
    }
  } else if (document.HasMember("url")) {
    const std::string& url = document["url"].GetString();
    if (url.find("wallet") != std::string::npos) {
      if (document.HasMember("data")){
        auto wallet_infos= document["data"].GetArray();
        for (const auto& wallet_info : wallet_infos) {
          std::string_view currency = wallet_info["currency"].GetString();
          for (const auto& product : upd_mgr_.products()) {
            if (product->base().native_currency() == currency) {
              Wallet* wallet = upd_mgr_.GetWalletMutable();
              UpbitFeedUpdateParser::ParseWallet(wallet_info, wallet);
              upd_mgr_.PublishWallet(product->feed_native_symbol());
            }
          }
        }
      }
    } else {
      size_t candles = url.find("candles/") + 8;
      size_t sep = url.find('?');
      std::string channel = url.substr(candles, sep - candles + 1);
      if (!CheckChannelFilter(channel)) {
        return;
      }

      auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
      std::string native_symbol = parameters["market"];
      fastfeed::proto::KlineInterval kinterval =
          UpbitFeedUpdateParser::ParseKlineInterval(channel);

      auto* kline = upd_mgr_.GetKlineMutable(native_symbol, kinterval);
      if (kline != nullptr) {
        int page = -1;
        if (parameters.count("page") > 0) {
          page = std::stoi(parameters["page"]);
        }
        const auto& kline_array = document["data"].GetArray();
        if (page == 0) {
          kline->SnapshotClear();
        } else {
          kline->UpdateClear();
        }
        bool to_publish = false;
        for (int i = kline_array.Size() - 1; i >= 0; i--) {
          auto& kline_json = kline_array[i];
          bool freezed = UpbitFeedUpdateParser::ParseKline(kline_json, kline, packet.timestamp,
              kinterval);
          to_publish = to_publish || freezed;
        }

        if (to_publish) upd_mgr_.PublishKline(native_symbol);
      }
    }
  } else {
    // meta msg like subscription response...
    LOG(INFO) << "[Upbit Feed Processor] meta msg for " << packet.topic_name << ": "
              << packet.payload.data();
  }

  upd_mgr_.onEndPacket(packet);
}  // namespace coin2::exchange::upbit::feed_v1::impl
}  // namespace coin2::exchange::upbit::feed_v1::impl
